package com.flink.demo.redis;

import com.kingstar.flink.start.BaseJob;
import com.kingstar.flink.start.TableFactory;
import com.kingstar.stream.table.BaseTable;
import org.apache.flink.streaming.api.datastream.DataStream;

/**
 * Redis异步查询demo  对来源值进行映射输出
 * <p>
 * 例redis 存储数据如下
 * key  value
 * 1    a
 * 2    b
 * 3    c
 * <p>
 * redis异步查询前后Stream结果
 * source   result
 * 1        1,a
 * 2        2,b
 * 3        3,c
 *
 * @author xin.ding
 * @date 2021/3/18 14:35
 */
public class RedisJoinJobDemo extends BaseJob {

    public static void main(String[] args) throws Exception {
        args = new String[]{"-t", "D:\\code\\library\\util\\utils-all\\flink\\flink-kingstar-demo\\src\\main\\java\\com\\flink\\demo\\redis\\redis-demo.json"};
        RedisJoinJobDemo job = new RedisJoinJobDemo();
        job.setJobName("test-redis");
        job.run(args);
    }

    @Override
    protected void execute() throws Exception {
        //构建Source Stream
        final DataStream rowSource = executor.createInput(new RowDataSource(), "row-source");
        rowSource.print("source");

        //读取json中redis配置项 redis_table 创建redis Table
        BaseTable redisTable = new TableFactory().build(getTableConfig("redis_table_standalone"));
        //redis 异步查询方法 使用方法同jdbc异步查询
        JoinRedisFunction joinRedisFunction = new JoinRedisFunction(redisTable);
        final DataStream joinRedis = executor.streamAsyncJoinProcess(rowSource, joinRedisFunction, "join_redis");

        //输出redis处理后结果流
        joinRedis.print("join");
    }

}

 /*
    执行结果：
    [2021-03-20 15:51:20.908][INFO][akka.event.slf4j.Slf4jLogger$$anonfun$receive$1.applyOrElse(Slf4jLogger.scala:92)] Slf4jLogger started
    [2021-03-20 15:51:20.967][INFO][akka.event.slf4j.Slf4jLogger$$anonfun$receive$1.applyOrElse(Slf4jLogger.scala:92)] Slf4jLogger started
    source:2> 2
    source:6> 2
    source:8> 2
    source:7> 2
    source:1> 1
    join:7> 2,b
    join:8> 1,a
    join:1> 2,b
    join:5> 2,b
    join:6> 2,b
  */